package cn.linjianhui.flink.sample;

import cn.linjianhui.flink.sample.model.OmsOrder;
import cn.linjianhui.flink.sample.util.ParamUtils;
import cn.linjianhui.flink.sample.util.Utils;
import com.alibaba.fastjson.JSON;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer011;
import org.apache.flink.streaming.connectors.redis.RedisSink;
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommand;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommandDescription;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisMapper;

import java.time.LocalDate;
import java.time.ZoneOffset;
import java.util.Properties;

public class PayCustomerJob {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = Utils.getStreamExecutionEnvironment();

        Properties props = ParamUtils.getKafkaConsumerProperty(args, "pay-customer", "earliest");

        DataStream<String> sourceStream = env.addSource(new FlinkKafkaConsumer011<>("OMS-Order", new SimpleStringSchema(), props));

        DataStream<String> payCustomerStream = sourceStream
                .map((MapFunction<String, OmsOrder>) val -> JSON.parseObject(val, OmsOrder.class))
                .filter((FilterFunction<OmsOrder>) order -> {
                    long currentDayTimestamp = LocalDate.now().atStartOfDay(ZoneOffset.ofHours(8)).toInstant().toEpochMilli();
                    return order.getORDERSTATUS() == 3 && order.getLASTMODIFYTIME() >= currentDayTimestamp;
                })
                .map((MapFunction<OmsOrder, String>) OmsOrder::getCUSCODE);

        FlinkJedisPoolConfig jedisPoolConfig = ParamUtils.getJedisPoolConfig(args);

        payCustomerStream.addSink(new RedisSink<>(jedisPoolConfig, new RedisMapper<String>() {
            @Override
            public RedisCommandDescription getCommandDescription() {
                return new RedisCommandDescription(RedisCommand.PFADD);
            }

            @Override
            public String getKeyFromData(String s) {
                return "flink-sample-customer";
            }

            @Override
            public String getValueFromData(String val) {
                return val;
            }
        }));

        env.execute("PayCustomerJob");
    }
}
